%% @doc Fast, write-once "tables" using flat files.
%% @end
-module (flatula).

-export ([ start/0,
	   stop/0,
	   open/2,
	   close/1,
	   read/2,
	   write/2,
	   write/3,
	   info/2 ]).

-behaviour (application).
-export ([ start/2,
	   stop/1 ]).

-ifdef (HAVE_EUNIT).
-include_lib ("eunit/include/eunit.hrl").
-export ([ spawn_benchmark/4 ]).
-export ([ graph_spawn_benchmark/3 ]).
-endif.

%
% public
%

%% @spec start () -> ok
%% @doc Start flatula.
%% @end

start () ->
  application:start (flatula).

%% @spec stop () -> ok
%% @doc Stop flatula.
%% @end

stop () ->
  application:stop (flatula).

%% @spec open (Table::any(), [Option]) -> { ok, Table } | { error, Reason }
%%       Option = { base_path, Path }
%%              | { max_file_bytes, Bytes }
%%              | { max_total_bytes, Bytes }
%%              | { n_partitions, NPartitions }
%% @doc Open the flatula table named Table. More than one process may open
%% a given table, and a single process may open it more than once; the table
%% is not closed until a matching close() is called for each open().
%%
%% The following options are mandatory:
%% <ul>
%% <li>base_path: the base path for files in the table, of the form
%% /PATH/BASENAME. flatula may create any files of the form
%% /PATH/BASENAME.EXTENSION.</li>
%% <li> max_file_bytes: maximum bytes in one table file. When an entry is
%% added that exceeds this limit, a new file is created.</li>
%% <li> max_total_bytes: maximum total bytes in the table. When an entry is
%% added that exceeds this limit, the oldest file in the table is removed
%% (along with all the entries in it).</li>
%% </ul>
%% The following options are optional:
%% <ul>
%% <li>n_partitions: number of server processes (default 1). Using more than
%% one partition can increase throughput, but is usually unnecessary.</li>
%% </ul>
%% @end

open (Table, Options) when is_list (Options) ->
  case is_list (proplists:get_value (base_path, Options)) andalso
       is_integer (proplists:get_value (max_file_bytes, Options)) andalso
       is_integer (proplists:get_value (max_total_bytes, Options))
  of
    true ->
      gen_server:call (flatula_server,
		       { open, Table, Options, self () },
		       infinity);
    false ->
      erlang:error ({ badarg, Options })
  end.

%% @spec close (Table::any()) -> ok
%% @doc Close the table Table. More than one process may open a given table,
%% and a single process may open it more than once; the table is not closed
%% until a matching close() is called for each open().
%% @end

close (Table) ->
  gen_server:call (flatula_server, { close, Table, self () }, infinity).

%% @type id(). An opaque identifier for entries in a table.

%% @spec read (any(), id()) -> { ok, any() } | not_found | { error, Reason }
%% @doc Read the entry with the given Id from the given Table.
%% Return { ok, Item } if found.
%% @end

read (Table, { Partition, Id }) when is_integer (Partition) ->
  Pids = flatula_server:table_pids (Table),
  if
    length (Pids) < Partition ->
      % TODO: should we store the number of partitions in the filesystem?
      not_found;
    length (Pids) >= Partition ->
      gen_server:call (lists:nth (Partition, Pids), { read, Id })
  end;
read (Table, Id) ->
  read (Table, { 1, Id }).

%% @spec write (any(), any()) -> Id::id()
%% @doc Write an entry to the given Table. Return an Id which may be used
%% to read the entry.
%% @end.

write (Table, Term) ->
  Pids = flatula_server:table_pids (Table),
  Partition = 1 + erlang:phash2 (now (), length (Pids)),
  Id = gen_server:call (lists:nth (Partition, Pids), { write, Term }, infinity),
  case Partition of
    1 -> Id;
    _ -> { Partition, Id }
  end.

%% @spec write (any(), any(), Timeout) -> { ok, Id::id() } | timeout
%%       Timeout = Milliseconds::integer() | infinity
%% @doc Write an entry to the given Table. Return an Id which may be used
%% to read the entry. If the given Timeout in milliseconds expires, the
%% atom 'timeout' is returned instead.
%% @end.

write (Table, Term, Timeout) when is_integer (Timeout), Timeout > 0
				; Timeout =:= infinity ->
  Pids = flatula_server:table_pids (Table),
  Partition = 1 + erlang:phash2 (now (), length (Pids)),
  Pid = lists:nth (Partition, Pids),
  try
    Id = gen_server:call (Pid, { write, Term }, Timeout),
    case Partition of
      1 -> { ok, Id };
      _ -> { ok, { Partition, Id } }
    end
  catch
    exit:{ timeout, { gen_server, call, [ Pid, _, _ ] } } -> timeout
  end.

%% @spec info (any(), What) -> [Value]
%%       What = size | max_file_bytes | max_total_bytes
%%       Value = { size, Bytes::integer }
%%             | { max_file_bytes, integer() }
%%             | { max_total_bytes, integer() }
%% @doc Get table information.
%% @end.

info (Table, What) ->
  [ gen_server:call (Pid, { info, What })
    || Pid <- flatula_server:table_pids (Table) ].

%
% application callbacks
%

%% @hidden
start (_Type, _Args) ->
  flatula_supervisor:start_link ().

%% @hidden
stop (_State) ->
  ok.

%
% tests
%

-ifdef (EUNIT).

setup_test_node () ->
  { ok, [ App ] } = file:consult ("../src/flatula.app"),
  ok = application:load (App),
  ok = flatula:start ().

teardown_test_node (Dir) ->
  flatula:stop (),
  application:unload (flatula),
  os:cmd ("rm -rf '" ++ Dir ++ "'").

make_setup () ->
  fun () -> setup_test_node () end.

make_teardown (Dir) ->
  fun (_) -> teardown_test_node (Dir) end.

simple_test_ () ->
  Dir = "tmp-" ++ os:getpid (),
  { setup,
    make_setup (),
    make_teardown (Dir),
    fun () ->
      { ok, foo } = flatula:open (foo, [ { base_path, Dir ++ "/foo" },
					 { max_file_bytes, 64 },
					 { max_total_bytes, 256 } ]),
      Id1 = flatula:write (foo, bar),
      Id2 = flatula:write (foo, baz),
      Id3 = flatula:write (foo, quux),
      { ok, bar } = flatula:read (foo, Id1),
      { ok, baz } = flatula:read (foo, Id2),
      { ok, quux } = flatula:read (foo, Id3),
      ok = flatula:close (foo),
      { ok, foo } = flatula:open (foo, [ { base_path, Dir ++ "/foo" },
					 { max_file_bytes, 64 },
					 { max_total_bytes, 256 } ]),
      { ok, bar } = flatula:read (foo, Id1),
      { ok, baz } = flatula:read (foo, Id2),
      { ok, quux } = flatula:read (foo, Id3),
      fill_table (foo),
      not_found = flatula:read (foo, Id1),
      ok = flatula:close (foo)
    end
  }.

timeout_test_ () ->
  Dir = "tmp-" ++ os:getpid (),
  { setup,
    make_setup (),
    make_teardown (Dir),
    fun () ->
      % a term which cannot possibly be written in 1 millisecond? :)
      Impossible = lists:duplicate (65536, 69),
      Table = { my, table },
      { ok, Table } = flatula:open (Table,
				    [ { base_path, Dir ++ "/t" },
				      { max_file_bytes, 64 * 1024 * 1024 },
				      { max_total_bytes, 256 * 1024 * 1024 } ]),
      timeout = flatula:write (Table, Impossible, 1),
      { ok, _Id } = flatula:write (Table, Impossible, 10000),
      ok = flatula:close (Table)
    end
  }.

multiple_partitions_test_ () ->
  Dir = "tmp-" ++ os:getpid (),
  { setup,
    make_setup (),
    make_teardown (Dir),
    fun () ->
      T = { my, table },
      BasePath = Dir ++ "/t",
      { ok, T } = flatula:open (T, [ { base_path, BasePath },
				     { n_partitions, 5 },
				     { max_file_bytes, 64 },
				     { max_total_bytes, 256 } ]),
      Id1 = flatula:write (T, bar),
      Id2 = flatula:write (T, baz),
      Id3 = flatula:write (T, quux),
      { ok, bar } = flatula:read (T, Id1),
      { ok, baz } = flatula:read (T, Id2),
      { ok, quux } = flatula:read (T, Id3),
      ok = flatula:close (T),
      { ok, T } = flatula:open (T, [ { base_path, BasePath },
				     { n_partitions, 5 },
				     { max_file_bytes, 64 },
				     { max_total_bytes, 256 } ]),
      { ok, bar } = flatula:read (T, Id1),
      { ok, baz } = flatula:read (T, Id2),
      { ok, quux } = flatula:read (T, Id3),
      fill_table (T),
      not_found = flatula:read (T, Id1),
      ok = flatula:close (T)
    end
  }.

fill_table (Table) ->
  Max = lists:sum ([ B || { max_total_bytes, B }
			    <- flatula:info (Table, max_total_bytes) ]),
  fill_table (Table, Max).

fill_table (Table, Max) ->
  flatula:write (Table, []),
  Bytes = lists:sum ([ B || { size, B } <- flatula:info (Table, size) ]),
  case Bytes < Max of
    true ->
      fill_table (Table, Max);
    false ->
      % XXX: the table is full, but we would have to guarantee a write to
      % every partition in order to drive out the 1st entry for sure
      lists:foreach (fun (_) -> flatula:write (Table, []) end,
		     lists:seq (1, 100)),
      ok
  end.

%
% benchmarking
%

spawn_benchmark (Sec, CPUs, Rs, NPs) ->
  Dir = "tmp-" ++ os:getpid (),
  Data = [ begin
	     Quantiles = spawn_benchmark (Dir, NP, MF, MT, S, R, WB, CPU),
	     { CPU, R, NP, Quantiles }
	   end
	   || NP <- NPs,
	      MF <- [ 8 * 1024 * 1024 ],
	      MT <- [ 256 * 1024 * 1024 ],
	      S <- [ Sec ],
	      R <- Rs,
	      WB <- [ 4096 ],
	      CPU <- CPUs ],
  error_logger:info_report ([ { data, lists:sort (Data) } ]),
  ok.

spawn_benchmark (Dir,
		 NPartitions,
		 MaxFileBytes,
		 MaxTotalBytes,
		 Seconds,
		 Rate,
		 WriteBytes,
		 CPU) ->
  Table = { my, table },
  SubDir = Dir ++ "/t",
  BasePath = SubDir ++ "/t",
  filelib:ensure_dir (BasePath),
  { ok, Table } = flatula:open (Table, [ { base_path, BasePath},
					 { max_file_bytes, MaxFileBytes },
					 { max_total_bytes, MaxTotalBytes },
					 { n_partitions, NPartitions } ]),
  Parent = self (),
  NSpawns = round (Seconds * Rate),
  spawn (fun () ->
	   Deadline = now_add (now (), round (1.0e6 * Seconds)),
	   T0 = now (),
	   spawn_loop (Parent, Table, WriteBytes, CPU, Deadline, NSpawns),
	   Elapsed = 1.0e-6 * timer:now_diff (now (), T0),
	   error_logger:info_report ([ { elapsed, Elapsed } ])
	 end),
  Quantiles = collect_stats (NSpawns, []),
  ok = flatula:close (Table),
  "ok" = os:cmd ("rm -r " ++ SubDir ++ " && printf ok"),
  Quantiles.

collect_stats (N, Acc) when N > 0 ->
  receive
    { write, DT } -> collect_stats (N - 1, [ DT | Acc ])
  end;
collect_stats (0, Acc) ->
  Sorted = lists:sort (Acc),
  L = length (Sorted),
  [ quantile (N/10, L, Sorted) || N <- lists:seq (1, 9) ].

quantile (Q, Len, List) ->
  F = Q * Len,
  T = trunc (F),
  case F == T of
    true  -> 1.0 * lists:nth (T, List);
    false -> 0.5 * (lists:nth (T, List) + lists:nth (T + 1, List))
  end.

spawn_loop (Parent, Table, Bytes, CPU, Deadline, N) when N > 0 ->
  Ms = round (timer:now_diff (Deadline, now ()) / N / 1000),
  if
    Ms > 0 -> timer:sleep (Ms);
    true   -> ok
  end,
  spawn (fun () -> one_shot_client (Parent, Table, Bytes, CPU) end),
  spawn_loop (Parent, Table, Bytes, CPU, Deadline, N - 1);
spawn_loop (_Parent, _Table, _Bytes, _CPU, _Deadline, 0) ->
  ok.

one_shot_client (Parent, Table, Bytes, CPU) ->
  Blob = compute_blob (Bytes, []),
  if
    CPU > 0 ->
      lists:foreach (fun (_) -> erlang:md5 (Blob) end, lists:seq (1, CPU));
    true ->
      ok
  end,
  T0 = now (),
  _ = flatula:write (Table, Blob),
  DT = timer:now_diff (now (), T0),
  Parent ! { write, DT }.

% make worker processes consume some cpu
compute_blob (Bytes, Acc) ->
  case iolist_size (Acc) < Bytes of
    true ->
      compute_blob (Bytes, [ erlang:md5 (term_to_binary (now ())) | Acc ]);
    false ->
      iolist_to_binary (Acc)
  end.

now_add ({ Mega, Sec, Micro }, Add) ->
  proper ({ Mega, Sec, Micro + Add }).

proper (Time = { _, Sec, Micro }) when Sec < 1000000, Micro < 1000000 ->
  Time;
proper ({ Mega, Sec, Micro }) when Sec < 1000000 ->
  proper ({ Mega, Sec + 1, Micro - 1000000 });
proper ({ Mega, Sec, Micro }) ->
  proper ({ Mega + 1, Sec - 1000000, Micro }).

% requires erlplot
% Data = [ { CPU, R, NP, Quantiles } ]
graph_spawn_benchmark (Basename, _EmulatorInfo, Data) ->
  AllParams = lists:usort ([ { CPU, R } || { CPU, R, _, _ } <- Data ]),
  Graphs =
    [ begin
	ParamData = [ { NP, Quantiles }
		      || { CPUx, Rx, NP, Quantiles } <- Data,
			 CPUx =:= CPU,
			 Rx =:= R ],
        GraphFile = fmt ("~s-~w-~w.png", [Basename, CPU, R]),
	Options = [ { title, fmt ("CPU ~w Rate ~w", [CPU, R]) } ],
	Plots =
	  [ { [ { title, integer_to_list (10 * N) ++ "%" } ],
	      [ { NP, lists:nth (N, Quantiles) }
		|| { NP, Quantiles } <- ParamData ] }
	    || N <- lists:seq (1, 9) ],
	ok = erlplot:plot (GraphFile, Options, Plots),
	{ GraphFile, CPU, R }
      end
      || { CPU, R } <- AllParams ],
  Graphs.

fmt (Fmt, Args) ->
  binary_to_list (iolist_to_binary (io_lib:format (Fmt, Args))).

-endif.
